package com.algo.panelmn;

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;  
import org.apache.hadoop.conf.Configured;  
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;  
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;  
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.*;
import org.apache.hadoop.mapreduce.lib.output.*;
import org.apache.hadoop.util.Tool;  
import org.apache.hadoop.util.ToolRunner;  

import com.algo.iplibrary.IpLibrary;
import com.supertool.tong.merger.util.MzSequenceFileCaidInputFormat;

public class Method2CookieUVStep1 extends Configured implements Tool {  
	
	public static  class UuidPartitioner extends Partitioner<UuidWritable, Text>{
		public int getPartition(UuidWritable key, Text value, int numPartitions) {
			return (int) (Math.abs(key.getUuid().hashCode())%numPartitions);
		}
	}
	public static class UuidSortComparator extends WritableComparator{
		public UuidSortComparator(){//WritableComparator must have this construction function
			super(UuidWritable.class, true);
		}
		public int compare(WritableComparable comkey1, WritableComparable comkey2) {
			UuidWritable key1 = (UuidWritable)comkey1;
			UuidWritable key2 = (UuidWritable)comkey2;
			if(key1 == null || key2 == null){	System.err.println("null key here: uuidsort"); return 1;}
			int cmp;
			if((cmp = key1.compareTo(key2)) != 0)	return cmp;
			else	return key1.compareTimeTo(key2);
		}
	}
	public static class UuidGroupSort extends WritableComparator{
		public UuidGroupSort(){//WritableComparator must have this construction function
			super(UuidWritable.class, true);
		}
		public int compare(WritableComparable comkey1, WritableComparable comkey2) {
			UuidWritable key1 = (UuidWritable)comkey1;
			UuidWritable key2 = (UuidWritable)comkey2;
			if(key1 == null || key2 == null){	System.err.println("null key here: uuidgroup"); return 1;}
			return key1.compareTo(key2);
		}
	}
	public static class Mappercamp extends org.apache.hadoop.mapreduce.Mapper<LongWritable, Text, UuidWritable, Text>{
		UuidWritable wrt = new UuidWritable();
		Set spids = new HashSet();
		boolean hasSpid = false;
		IpLibrary iplib = null;
		public void setup(Context context) throws IOException{
			Configuration conf = context.getConfiguration();
			if(conf.get("iplibrary") != null){
				iplib = new IpLibrary(conf.get("iplibrary"), conf.get("filesystem"));
				System.out.print("iplib not null");
			}
			else System.out.print("iplib null");
			String spidfilename;
			if((spidfilename = context.getConfiguration().get("spidfilename")) != null){
				FileReader fr = null;
				BufferedReader bf = null;
				try {
					fr = new FileReader(spidfilename);
					bf = new BufferedReader(fr);
					String line;
					while((line = bf.readLine()) != null)
						spids.add(line);
					hasSpid = true;
				} catch (FileNotFoundException e) {
					System.out.println("don't have spid file");
				}
				finally{
					if(bf != null) bf.close();
					if(fr != null) fr.close();
				}
			}
			System.out.println(spids.toString());
		}
		//read campaign from ETL log file, output ((mzid, time), "0")
		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String line = value.toString();
			String[] items = line.split("[=\\^]");
			Map map = new HashMap();
			for(int i = 1; i < items.length; i = i + 2)
				map.put(items[i-1], items[i]);
			try{
				int ispc = Integer.parseInt((String)map.get("plt"));
				if(ispc == 0){
					String stable = (String) map.get("av");
					String type = (String) map.get("tp");
					if(type.equals("imp") && (stable.equals("10") || stable.equals("11"))){
						String uuid = (String) map.get("uuid");
						long time = Long.parseLong((String) map.get("ti"));
						wrt.set(uuid, time);
						
						String location;
						if(iplib != null){
							String ip = (String) map.get("ip");
							location = iplib.getIPLocation(ip) + "^loc";
						}
						else	location = "0^loc";
						
						if(hasSpid){
							String spid = (String) map.get("p");
							if(spids.contains(spid))
								context.write(wrt, new Text(location));
						}
						else{
							System.out.println("no spid.txt file found, the result didnot filter with specific spids");
							context.write(wrt, new Text(location));
						}
					}
				}
			}catch(NullPointerException e){
				System.out.println("this line do not contain necessary items, was aborted");
			}
		}
	}
	public static class Mappersns extends org.apache.hadoop.mapreduce.Mapper<LongWritable, Text, UuidWritable, Text> {
		//read cookie sns mapping file, output ((mzid, time), snsid)
		UuidWritable wrt = new UuidWritable();
		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String str = value.toString();
			String[] sns = str.split("\\^");
			if(sns.length >= 4){
				String uuid = sns[0];
				String snsid = sns[1];
				long time = Long.parseLong(sns[2]);
				wrt.set(uuid, time);
				context.write(wrt, new Text(snsid));
			}
			else	System.out.println("less sns element");
		}
	}
	public static class MapperPanel extends Mapper<LongWritable, Text, UuidWritable, Text>{
		UuidWritable wrt = new UuidWritable();
		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String str = value.toString();
			int pos = str.indexOf(",");
			if(pos < 0) return;
			wrt.set(str.substring(0, pos), -1);
			String[] items = str.substring(pos+1).split(",");
			int sex = 0, age = 0, edu = 0, salary = 0;
			if(items[3].equals("1"))	sex = 1;
			for(int i = 5; i <= 11; i++)
				if(items[i].equals("1"))	age = i - 4;
			for(int i = 13; i <= 15; i++)
				if(items[i].equals("1"))	edu = i - 12;
			for(int i = 17; i <= 20; i++)
				if(items[i].equals("1"))	salary = i - 16;
			context.write(wrt, new Text(sex + "," + age + "," + edu + "," + salary));
		}
	}
	private static String parseOrgPanel(String[] items){
		int sex = 0, age = 0, edu = 0, salary = 0;

	    if(items[2].equals("3")) sex = 1;
	    
	    if(items[3].equals("6")) age = 1;
	    else if(items[3].equals("7")) age = 2;
	    else if(items[3].equals("8")) age = 3;
	    else if(items[3].equals("9")) age = 4;
	    else if(items[3].equals("10")) age = 5;
	    else if(items[3].equals("11")) age = 6;
	    else if(items[3].equals("41")) age = 7;
	    
	    if(items[4].equals("17")) edu = 1;
	    else if(items[4].equals("18")) edu = 2;
	    else if(items[4].equals("42")) edu = 3;
	    
	    if(items[5].equals("22")) salary = 1;
	    else if(items[5].equals("23")) salary = 2;
	    else if(items[5].equals("24")) salary = 3;
	    else if(items[5].equals("25")) salary = 4;
		
	    return new StringBuilder().append(sex).append(",").append(age).append(",").append(edu).append(",").append(salary).toString();
	}
	public static class MapperOrgPanel extends Mapper<LongWritable, Text, UuidWritable, Text>{
		UuidWritable wrt = new UuidWritable();
		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String str = value.toString();
			int pos = str.indexOf(",");
			if(pos < 0)	return;
			wrt.set(str.substring(0, pos), -1);
			String[] items = str.substring(pos+1).split(",");
			context.write(wrt, new Text(parseOrgPanel(items)));
		}
	}
	public static class MapperCookieTA extends Mapper<LongWritable, Text, UuidWritable, Text>{
		UuidWritable wrt = new UuidWritable();
		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String str = value.toString();
			String uuidTA = str.substring(1, str.length()-1);
			int pos = uuidTA.indexOf(",");
			wrt.set(uuidTA.substring(0, pos), 0);
			context.write(wrt, new Text(uuidTA.substring(pos+1)));
		}
	}
	public static class Reducer extends org.apache.hadoop.mapreduce.Reducer<UuidWritable, Text, Text, Text>{
		private MultipleOutputs mos;
		String uuid_pv, uuid_sns, uuid_sns_pv;
		public void setup(Context context){
			mos = new MultipleOutputs(context);
			uuid_pv = "uuid_pv/pv";
			uuid_sns = "uuid_sns/sns";
			uuid_sns_pv = "uuid_sns_pv/sns_pv";
		}
		public void cleanup(Context context) throws IOException, InterruptedException{
			mos.close();
		}
		//the same uuid was passed to the same reducer and sorted in time
		//tag the campaign mzid with appropriate snsid, output (snsid, pv) and (mzid, snsid) mapping
		public void reduce(UuidWritable key, Iterable<Text> values,Context context)
			 throws IOException, InterruptedException {
			long psnsnum = 0;
			String psnsid = null;//the prior sns id
			String panel = null; //save panel
			Map sns = new HashMap();
			Map loc = new HashMap();
			Iterator it = values.iterator();
			while(it.hasNext()){
				String lw = ((Text)it.next()).toString();
				if(key.getTime() <= 0){//this uuid has corresponding record in panel
					if(panel == null)	panel = lw;
					continue;
				}
				String[] items = lw.split("\\^");
				if(items.length == 1){
					psnsid = lw;
					if(psnsnum > 0) sns.put(psnsid, new Counter(psnsnum));
					break;		//break at the first sns id
				}
				else{
					psnsnum++;
					if(!items[0].equals("0"))
						if(loc.containsKey(items[0])) ((Counter)loc.get(items[0])).num++;
						else loc.put(items[0], new Counter(1));
						//mos.write("uuid", new Text(items[0] + "^" + key.getUuid()), new Text("pv-" + 1), uuid_pv);
				}
			}
			while(it.hasNext()){
				String lw = ((Text)it.next()).toString();
				String[] items = lw.split("\\^");
				if(items.length == 2){//campaign
					psnsnum++;
					if(psnsid != null){
						if(sns.containsKey(psnsid)) ((Counter)sns.get(psnsid)).num++;
						else sns.put(psnsid, new Counter(1));
					}
					if(!items[0].equals("0"))
						if(loc.containsKey(items[0])) ((Counter)loc.get(items[0])).num++;
						else loc.put(items[0], new Counter(1));
						//mos.write("uuid", new Text(items[0] + "^" + key.getUuid()), new Text("pv-" + 1), uuid_pv);
				}
				else	psnsid = lw;
			}
			if(psnsnum > 0 && sns.size() > 0){
				if(loc.size() > 0){
					it = loc.keySet().iterator();
					while(it.hasNext()){
						String location = (String) it.next();
						mos.write("uuid", new Text(key.getUuid()), new Text(((Counter)loc.get(location)).num + "^" + location), uuid_pv);
						if(panel != null)
						  mos.write("uuid", new Text(key.getUuid()), new Text(((Counter)loc.get(location)).num + "^" + location + "," + panel), uuid_pv);
					}
				}
				mos.write("uuid", new Text(key.getUuid()), new Text(psnsnum + "^total"), uuid_pv);
				if(panel != null)
					mos.write("uuid", new Text(key.getUuid()), new Text(psnsnum + "^total" + "," + panel), uuid_pv);
				it = sns.keySet().iterator();
				while(it.hasNext()){
					String snsid = (String) it.next();
					mos.write("uuid", new Text(key.getUuid()), new Text(snsid), uuid_sns);
					if(uuid_sns_pv != null)
						mos.write("uuid", new Text(key.getUuid()), new Text(snsid + ":" + ((Counter)sns.get(snsid)).num), uuid_sns_pv);
				}
			}
		}
	}
	public int getMonthDay(int year, int month){
		if(month > 12 || month < 1) return 0;
		if(month%2 == 1 && month <= 7) return 31;
		else if(month%2 == 0 && month >= 8) return 31;
		else if(month != 2) return 30;
		else{
			if((year%4 == 0 && year%100 != 0) || year%400 == 0) return 29;
			else return 28;
		}
	}
    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        conf.set("caidfilename", "caid.txt");
        conf.set("spidfilename", "spid.txt");
        String panel = null, cookieTA = null;
        if(args.length <= 5)
	    	{ System.out.print("parameter less than 6");	return -1;}
	    else if(args.length == 8){
	    	conf.set("iplibrary", args[6]);
	    	conf.set("filesystem", args[7]);
	    }
	else if(args.length >= 10){
                conf.set("iplibrary", args[6]);
                conf.set("filesystem", args[7]);
		panel = args[8];
		cookieTA = args[9];
	}
	    else if(args.length >= 12)
	    	return -1;
        Job job = new Job(conf, "cookie-Step1");
        job.setJarByClass(Method2CookieUVStep1.class);
        
        int numReducer = 0;
        String type = args[2];
        FileSystem hdfs = FileSystem.get(conf);
        int fromyear = Integer.parseInt(args[3].substring(0, 4));
        int toyear = Integer.parseInt(args[4].substring(0, 4));
        int frommonth = Integer.parseInt(args[3].substring(4, 6));
        int tomonth = Integer.parseInt(args[4].substring(4, 6));
        int fromday = Integer.parseInt(args[3].substring(6));
        int today = Integer.parseInt(args[4].substring(6));
        for(int i = fromyear; i <= toyear; i++){
		int tomonth_y = tomonth;
                if(i != toyear) tomonth_y = 12;
                int frommonth_y = frommonth;
                if(fromyear != i)       frommonth_y = 1;
        	for(int j = frommonth_y; j <= tomonth_y; j++){
        		int lastday = getMonthDay(i, j);
        		int firstday = 1;
        		if(i == fromyear && j == frommonth)
        			firstday = fromday;
        		if(i == toyear && j == tomonth)
        			lastday = today;
        		String month = j < 10 ? ("0" + j) : ("" + j);
        		for(int k = firstday; k <= lastday; k++){
        			numReducer++;
        			String day = k < 10 ? ("0" + k) : ("" + k);
        			Path p1 = new Path(args[0] + "/" + i + month + day);
        			FileStatus[] files = hdfs.listStatus(p1);
        	        for(int l = 0;l < files.length;l++){
        	        	Path file = files[l].getPath();
        	        	if(file.getName().indexOf(type) >= 0)
        	        		MultipleInputs.addInputPath(job, file, MzSequenceFileCaidInputFormat.class, Mappercamp.class);
        	        }
        		}
        	}
        }
        for(int i = fromyear; i <= toyear; i++){
                int tomonth_y = tomonth;
                if(i != toyear) tomonth_y = 12;
                int frommonth_y = frommonth;
                if(fromyear != i)       frommonth_y = 1;
        	for(int j = frommonth_y; j <= tomonth_y; j++){
        		int lastday = getMonthDay(i, j);
        		int firstday = 1;
        		if(i == fromyear && j == frommonth)
        			firstday = fromday;
        		if(i == toyear && j == tomonth)
        			lastday = today;
        		String month = j < 10 ? ("0" + j) : ("" + j);
			
			if(panel != null && cookieTA != null){
				if(hdfs.exists(new Path(cookieTA + "/" + i + month))){
                			MultipleInputs.addInputPath(job, new Path(panel + "/l" + i + month), TextInputFormat.class, MapperOrgPanel.class);
                			MultipleInputs.addInputPath(job, new Path(cookieTA + "/" + i + month + "/*"), TextInputFormat.class, MapperCookieTA.class);
				}
        		}
        		for(int k = firstday; k <= lastday; k++){
        			String day = k < 10 ? ("0" + k) : ("" + k);
        			Path p2 = new Path(args[1] + "/" + i + month + "/" + i + month + day);
        			MultipleInputs.addInputPath(job, p2, TextInputFormat.class, Mappersns.class);
        		}
        	}
        }
        
        Path pout = new Path(args[5]);
        FileOutputFormat.setOutputPath(job, pout);
        job.setPartitionerClass(UuidPartitioner.class);
        job.setGroupingComparatorClass(UuidGroupSort.class);
        job.setSortComparatorClass(UuidSortComparator.class);
        
        job.setReducerClass(Reducer.class);
        job.setMapOutputKeyClass(UuidWritable.class);
        job.setMapOutputValueClass(Text.class);
        
        if(RunChainJob2.isBatchCaids)	job.setNumReduceTasks(RunChainJob2.step1ReducerNum);
	else	job.setNumReduceTasks(3 + numReducer/5);
        
        MultipleOutputs.addNamedOutput(job, "uuid", TextOutputFormat.class, Text.class, Text.class);
        return job.waitForCompletion(true) ? 1 : 0;
    }
    public static void main(String[] args) throws Exception {
        int res = ToolRunner.run(new Configuration(), new Method2CookieUVStep1(), args);
        System.exit(res);
    }
}
